package com.chenjl.comsumer;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.chenjl.Constants;
/**
 * 消费者
 * KafkaConsumer is not safe for multi-threaded access
 * new Client API Producer,Comsumer 都直连Broker，由Broker和zk维护comsumer的offset
 * 2016-9-13 16:45:50
 * @author chenjinlong
 */
public class Consumer02 {
	private static final Logger logger = LoggerFactory.getLogger(Consumer02.class);
	
	public static void main(String[] args) {
		Properties props = new Properties();
		props.put("bootstrap.servers",Constants.BROKER_URL);
		props.put("client.id","1100");
		props.put("group.id","test consumer");
		props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
		
		Consumer<String, String> consumer = new KafkaConsumer<String, String>(props);
		
		consumer.subscribe(Arrays.asList(Constants.TOPIC_NAME));
		logger.info("consumer poll begin");
		ConsumerRecords<String, String> messages = consumer.poll(5000);
		logger.info("consumer poll end , message count : {}",messages.count());
		
		for(ConsumerRecord<String, String> message : messages) {
			logger.info("consumer message , key : {}, value : {} , offset : {}",message.key(),message.value(),message.offset());
		}
		
		logger.info("consumer end");
		//consumer.commitSync();
		consumer.close();
	}
}